package com.mqtt.demo1.service;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

@Slf4j
@Component
public class MqttService {

    @Autowired
    private MqttClient mqttClient;

    /**
     * 发送消息
     *
     * @param topic 主题
     * @param data  消息内容
     */
    public void sendMessage(String topic, Object data) {
        ObjectMapper mapper = new ObjectMapper();
        try {
            // 转换消息为json字符串
            String json = mapper.writeValueAsString(data);
            mqttClient.publish(topic, new MqttMessage(json.getBytes(StandardCharsets.UTF_8)));
        } catch (JsonProcessingException e) {
            log.error(String.format("MQTT: 主题[%s]发送消息转换json失败", topic));
        } catch (MqttException e) {
            log.error(String.format("MQTT: 主题[%s]发送消息失败", topic));
        }
    }

    /**
     * 订阅主题
     *
     * @param topic    主题
     * @param listener 消息监听处理器
     */
    public void subscribe(String topic, IMqttMessageListener listener) {
        try {
            mqttClient.subscribe(topic, listener);
        } catch (MqttException e) {
            log.error(String.format("MQTT: 订阅主题[%s]失败", topic));
        }
    }

}
